package com.pxene.dmp.reducer;

import java.io.BufferedReader;
import java.io.FileReader;
import java.io.IOException;
import java.net.URI;
import java.util.Date;
import java.util.HashMap;
import java.util.Map;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Reducer;

import com.pxene.dmp.constant.BaseConstant;

public class CountFavouriteSiteReducer extends Reducer<Text, LongWritable, LongWritable, Text> {
	private Log log = LogFactory.getLog(CountFavouriteSiteReducer.class);
	private LongWritable resultKey = new LongWritable();
	private Text resultVal = new Text();
	private Map<String, String> cache;

	@Override
	protected void setup(Context context) throws IOException, InterruptedException {
		Configuration conf = context.getConfiguration();
		// 设置配置文件名
		System.setProperty("config-file", conf.get("config-file"));

		URI[] uris = context.getCacheFiles();
		if (uris != null) {
			// 调优为128
			cache = new HashMap<String, String>(128);
			BufferedReader br = null;
			for (URI uri : uris) {
				String filename = uri.toString().substring(uri.toString().lastIndexOf("/") + 1);
				log.info(BaseConstant.LOG_PREFIX + "cache file is " + filename);
				// 优化为20M
				br = new BufferedReader(new FileReader(filename), 1024 * 20);
				String line = null;
				while ((line = br.readLine()) != null) {
					String[] values = line.split("\\|");
					if (values.length == 5) {
						cache.put(values[0] + "|" + values[1] + "|" + values[3] + "|" + values[4], values[2]);
					}
				}
				br.close();
			}
		}
	}

	@Override
	protected void reduce(Text key, Iterable<LongWritable> values, Context context) throws IOException, InterruptedException {
		String[] columns = key.toString().split("\\|");
		if (columns.length != 4) {
			return;
		}
		String imsi = columns[0];
		String site = columns[1];
		String province = columns[2];
		String dataType = columns[3];

		long count = 0l;
		for (LongWritable value : values) {
			count += value.get();
		}
		
		long sum = 0l;
		String keyStr = imsi + "|" + site + "|" + province + "|" + dataType;
		if (cache.containsKey(keyStr)) {
			sum = count + Long.parseLong(cache.get(keyStr));
		} else {
			sum = count;
		}

		resultKey.set(new Date().getTime());
		resultVal.set(imsi + "|" + site + "|" + sum + "|" + province + "|" + dataType);
		context.write(resultKey, resultVal);
	}

}
